package com.chat.service.impl;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.chat.config.NettyConfig;
import com.chat.domain.ChatGroup;
import com.chat.request.MessageGroupRequest;
import com.chat.request.SendMessageRequest;
import com.chat.request.WithdrawOneRequest;
import com.chat.service.ChatGroupService;
import com.chat.service.PushService;
import com.chat.utils.ExecutorServiceUtil;
import com.chat.utils.RedisMessageUtil;
import com.chat.response.message.WithdrawMessagesResponse;
import com.chat.response.message.GroupMessagesResponse;
import com.chat.response.message.ReceiveMessagesResponse;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Service
public class PushServiceImpl implements PushService {
    @Resource
    RedisMessageUtil redisMessageUtil;
    @Resource
    ChatGroupService chatGroupService;


    @Override
    public Boolean pushMsgToOne(SendMessageRequest sendMessageRequest) {
        Channel channel = getChannelForUser(sendMessageRequest.getToUserId());
        ReceiveMessagesResponse messageResponse = createReceiveMessageResponse(sendMessageRequest);
        if (channel == null) {
            saveOfflineMessage(sendMessageRequest.getToUserId(), messageResponse);
            return false;
        }
        sendOnlineMessage(channel, messageResponse);
        return true;
    }

    private Channel getChannelForUser(String userId) {
        return NettyConfig.getUserChannelMap().get(userId);
    }

    private ReceiveMessagesResponse createReceiveMessageResponse(SendMessageRequest request) {
        ReceiveMessagesResponse response = new ReceiveMessagesResponse();
        response.setMessage(request.getMessage());
        response.setToUserId(request.getToUserId());
        response.setFromUserId(request.getFromUserId());
        response.setMessageDate(DateUtil.now());
        response.setUsername(request.getUsername());
        response.setAvatar(request.getAvatar());
        response.setType(request.getType());
        response.setId(request.getId());
        return response;
    }

    private void saveOfflineMessage(String userId, ReceiveMessagesResponse messageResponse) {
        redisMessageUtil.addOfflineMessage(Long.valueOf(userId), messageResponse);
    }

    private void sendOnlineMessage(Channel channel, ReceiveMessagesResponse messageResponse) {
        channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(messageResponse)));
    }


    @Override
    public void pushMsgToAll(String msg) {
        NettyConfig.getChannelGroup().writeAndFlush(new TextWebSocketFrame(msg));
    }

    @Override
    public void pushMsgToGroup(MessageGroupRequest dto) {
        ChatGroup group = chatGroupService.getById(dto.getGroupId());
        List<String> recipientIds = getRecipientIds(group.getUserIds(), dto.getMessageUserId());
        ConcurrentHashMap<String, Channel> userChannelMap = NettyConfig.getUserChannelMap();

        List<Callable<Void>> tasks = createPushTasks(dto, recipientIds, userChannelMap);
        ExecutorServiceUtil.executeAndWaitForResults(tasks);
    }

    private List<String> getRecipientIds(List<String> userIds, String messageUserId) {
        List<String> recipientIds = new ArrayList<>(userIds);
        recipientIds.remove(messageUserId);
        return recipientIds;
    }

    private List<Callable<Void>> createPushTasks(MessageGroupRequest dto, List<String> recipientIds, ConcurrentHashMap<String, Channel> userChannelMap) {
        return recipientIds.stream()
                .map(id -> (Callable<Void>) () -> {
                    pushMessageToUser(dto, id, userChannelMap.get(id));
                    return null;
                })
                .collect(Collectors.toList());
    }

    private void pushMessageToUser(MessageGroupRequest dto, String userId, Channel channel) {
        GroupMessagesResponse vo = createGroupMessageResponse(dto, userId);
        if (Objects.isNull(channel)) {
            redisMessageUtil.addOfflineMessage(Long.valueOf(userId), vo);
        } else {
            channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(vo)));
        }
    }

    private GroupMessagesResponse createGroupMessageResponse(MessageGroupRequest dto, String userId) {
        GroupMessagesResponse vo = new GroupMessagesResponse();
        vo.setId(dto.getId());
        vo.setFromUserId(dto.getGroupId());
        vo.setMessageUserId(userId);
        vo.setUsername(dto.getUsername());
        vo.setMessage(dto.getMessage());
        vo.setMessageDate(DateUtil.now());
        vo.setMessageUser(dto.getMessageUser());
        vo.setAvatar(dto.getAvatar());
        vo.setType(dto.getType());
        return vo;
    }

    @Override
    public void withdrawOne(WithdrawOneRequest dto) {
        if (isGroupMessage(dto.getToUserId())) {
            withdrawGroupMessage(dto);
        } else {
            withdrawPrivateMessage(dto);
        }
    }

    private boolean isGroupMessage(String toUserId) {
        return toUserId.charAt(0) == 'G';
    }

    private void withdrawGroupMessage(WithdrawOneRequest dto) {
        ChatGroup group = chatGroupService.getById(dto.getToUserId());
        List<String> recipientIds = getRecipientIds(group.getUserIds(), dto.getUserId());
        WithdrawMessagesResponse response = createWithdrawResponse(dto.getMessageId());

        List<Callable<Void>> tasks = createWithdrawTasks(recipientIds, response);
        ExecutorServiceUtil.executeAndWaitForResults(tasks);
    }

    private WithdrawMessagesResponse createWithdrawResponse(String messageId) {
        WithdrawMessagesResponse response = new WithdrawMessagesResponse();
        response.setId(messageId);
        response.setType(3);
        return response;
    }

    private List<Callable<Void>> createWithdrawTasks(List<String> recipientIds, WithdrawMessagesResponse response) {
        return recipientIds.stream()
                .map(id -> (Callable<Void>) () -> {
                    sendWithdrawMessage(id, response);
                    return null;
                })
                .collect(Collectors.toList());
    }

    private void withdrawPrivateMessage(WithdrawOneRequest dto) {
        WithdrawMessagesResponse response = createWithdrawResponse(dto.getMessageId());
        sendWithdrawMessage(dto.getToUserId(), response);
    }

    private void sendWithdrawMessage(String userId, WithdrawMessagesResponse response) {
        Channel channel = NettyConfig.getUserChannelMap().get(userId);
        if (Objects.isNull(channel)) {
            redisMessageUtil.addOfflineMessage(Long.valueOf(userId), response);
        } else {
            channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(response)));
        }
    }
}

